package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo6Burk {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      //local 本地运行，  local[4]: 多线程运行
      .master("local")
      .config("spark.sql.shuffle.partitions", 1) // 设sql shuffle之后分区数, 默认是200
      .appName("burk")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    //1、读取数据
    val burksDF: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("burk STRING,year STRING,tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE,tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE,tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
      .load("spark/data/burks.txt")


    /**
      *
      * 1、统计每个公司每年按月累计收入
      *
      * 输出结果
      * 公司代码,年度,月份,当月收入,累计收入
      */

    /**
      *
      * sql
      */

    burksDF.createOrReplaceTempView("burk")


    spark.sql(
      """
        |select burk,year,month,tsl , sum(tsl) over(partition by burk ,year  order by month) as leji
        |from (
        |select burk,year,month,tsl from burk
        |LATERAL VIEW explode(map('01',tsl01,'02',tsl02,'03',tsl03,'04',tsl04,'05',tsl05,'06',tsl06,'07',tsl07,'08',tsl08,'09',tsl09,'10',tsl10,'11',tsl11,'12',tsl12)) t as month,tsl
        |) as a
        |
      """.stripMargin).show(100000)

    /**
      *
      * DSL
      *
      */


    val m: Column = map(
      expr("01"), $"tsl01",
      expr("02"), $"tsl02",
      expr("03"), $"tsl03",
      expr("04"), $"tsl04",
      expr("05"), $"tsl05",
      expr("06"), $"tsl06",
      expr("07"), $"tsl07",
      expr("08"), $"tsl08",
      expr("09"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )


    burksDF
      .select($"burk", $"year", explode(m) as Array("month", "tsl"))
      .select($"burk", $"year", $"month", $"tsl", sum($"tsl") over Window.partitionBy($"burk", $"year").orderBy($"month") as "leiji")
      .show(10000)

    /**
      *
      * 2、统计每个公司当月比上年同期增长率
      *
      * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
      *
      *
      * coalesce :返回第一个不为空的列
      *
      */

    burksDF
      .select($"burk", $"year", explode(m) as Array("month", "tsl"))
      .select($"burk", $"year", $"month", $"tsl", lag($"tsl", 1, 0.0) over Window.partitionBy($"burk", $"month").orderBy($"year") as "last")
      .select($"burk", $"year", $"month", $"tsl", $"last", $"tsl" / $"last" - 1 as "ps")
      .select($"burk", $"year", $"month", $"tsl", $"last", round(coalesce($"ps", expr("0.0")), 6))

      .show(10000)
  }

}
